package SparkGraphXInAction

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.graphx._
import org.apache.spark.graphx
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

/**
  * Created by Administrator on 2017/4/26 0026.
  */
object PreBySVDPlusPlus {
  def pred(g:Graph[(Array[Double],Array[Double],Double,Double),Double], mean:Double, u:Long, i:Long)={
    val user = g.vertices.filter(_._1 == u).collect()(0)._2
    val item = g.vertices.filter(_._1 == i).collect()(0)._2
    mean + user._3 + item._3 +
      item._1.zip(user._2).map(x => x._1*x._2).reduce(_ + _)
  }

  def main(args: Array[String]): Unit = {
    // 屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    //构建EdgeRDD(边RDD)，其实也可以从Graph中获取。
    //这个EdgeRDD，对应图7.2
    val edges = sc.makeRDD(Array(Edge(1L,11L,5.0),Edge(1L,12L,4.0),Edge(2L,12L,5.0),
      Edge(2L,13L,5.0),Edge(3L,11L,5.0),Edge(3L,13L,2.0),
      Edge(4L,11L,4.0),Edge(4L,12L,4.0)))
    //算法的指定超参数（参考表7.1）
    val svdconf = new lib.SVDPlusPlus.Conf(2, 10, 0, 5, 0.007, 0.007, 0.005, 0.015)
    //运行SVD++算法，获得返回的模型——输入图的再处理结果和数据集的平均打分情况。
    val (g,mean) = lib.SVDPlusPlus.run(edges,svdconf)
    //预测,在TestGraphIsomorphism.scala中有pred函数的另一种实现方式。
    //为什么要实现pred的另一种实现方式。因为我们想在map()的括号内使用pred，而在这种情况下，我们就不可以传递rdd或者图对象给pred了。
    //所以我们要使用另外一种pred的实现方式。
    val prediction = pred(g, mean, 4L, 13L)
    print(prediction)
    sc.stop()
  }
}
